package net.bwie.flink;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 使用Flink CDC 实时采集业务数据，发送Kafka消息队列
 * @author xuanyu
 * @date 2025/10/28
 */
public class _02GmallCdcJob {

	public static void main(String[] args) throws Exception {
		// 1. 执行环境-env
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1) ;


		// 2. 数据源-source
		/*
		https://nightlies.apache.org/flink/flink-cdc-docs-release-3.5/docs/connectors/flink-sources/mysql-cdc/#datastream-source
		 */
		env.enableCheckpointing(3000);
		MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
			.hostname("node101")
			.port(3306)
			.databaseList("flink_day12")
			.tableList(
				"flink_day12.comment_info", "flink_day12.payment_info",
				"flink_day12.order_info", "flink_day12.order_detail"
			)
			.username("root")
			.password("123456")
			// 全量+增量
			.startupOptions(StartupOptions.initial())
			.deserializer(new JsonDebeziumDeserializationSchema())
			.build();
		DataStreamSource<String> stream = env.fromSource(
			mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source"
		);

		// 4.数据接收器-sink
		KafkaSink<String> sink = KafkaSink.<String>builder()
			.setBootstrapServers("node101:9092,node102:9092,node103:9092")
			.setRecordSerializer(KafkaRecordSerializationSchema.builder()
				.setTopic("db-topic")
				.setValueSerializationSchema(new SimpleStringSchema())
				.build()
			)
			.setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
			.build();
		stream.sinkTo(sink);

		// 5. 触发执行-execute
		env.execute("GmallCdcJob") ;
	}

}
